package ubc.master;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import ubc.CrawlResult;
import ubc.CrawlWrapper;
import ubc.Queue;
import ubc.QueueOrganizer;

/* Server:
 * - Receives CrawlResult objects from clients
 * - Sends a list of uncrawled hostnames to the client
 * - Fully implements non-blocking io
 */

public class Server extends Thread{

    Selector selector;
    ServerSocketChannel server;
    QueueOrganizer qo;
    int _mode, _numSend, _time;
    Boolean _quit;
    long startTime, curTime, lastPrintTime;
    Date startDate;
    int numWorkers=0;
    List<String> workerNames = new ArrayList<String>();

    public Server(String bootStrapAddr, int bootStrapPort, int serverPort, int mode, int numSend, int time) throws Exception {
        _mode = mode;
        _numSend = numSend;
        _time = time;
        startTime = System.currentTimeMillis();
        startDate = new Date();
        createServer(serverPort);
        qo = new QueueOrganizer("Server", ubc.Constants.serverFilePath);
        qo.uncrawledHosts.enqueue(bootStrapAddr+":"+bootStrapPort);
        qo.start();
    }

    public void createServer(int port) throws Exception {

        // Create a new InetSocketAddress
        InetAddress localhost = InetAddress.getLocalHost();
        InetSocketAddress isa = new InetSocketAddress(localhost, port);
        System.out.println("Server started on: " + isa.getAddress() + ":" + isa.getPort());

        // Create a new Selector
        selector = Selector.open();

        // Register a new server channel to the selector
        server = ServerSocketChannel.open();
        server.configureBlocking(false);
        server.socket().bind(isa);
        server.register(selector, SelectionKey.OP_ACCEPT, "server");
    }

    public void run() {
        while(!Thread.interrupted()) {

            curTime = System.currentTimeMillis();

            if( (curTime - startTime) > _time*1000 ) {
                Thread.interrupted();
                break;
            }

            if ( (curTime - lastPrintTime) > 5000 ) {
                lastPrintTime = curTime;
                Calendar cal = Calendar.getInstance();
                System.out.println(cal.get(Calendar.HOUR_OF_DAY)+":"+cal.get(Calendar.MINUTE)+":"+cal.get(Calendar.SECOND) +
                        " Connections:" + selector.keys().size() +
                        " Workers:" + numWorkers +
                        " Uncrawled:" + qo.uncrawledHosts.size());
            }

            try {
                // Wait for an event
                selector.select(500);
            } catch (Exception e) {
                System.err.println("run.select.Exception");
                //e.printStackTrace();
            }
            Set keys = selector.selectedKeys();
            Iterator i = keys.iterator();
            while(i.hasNext() && !Thread.interrupted()) {

                SelectionKey key = (SelectionKey)i.next();
                i.remove();
                if( !key.isValid() ) {
                    continue;
                }
                if(key.isAcceptable()) {
                    accept(key);
                }
                else if(key.isReadable()) {
                    readFromClient(key);
                }
                else if(key.isWritable()) {
                    writeToClient(key, _mode);
                }
            }
        }

        try {
            for(int i=0; i<selector.keys().size(); i++) {
                Set<SelectionKey> sk = selector.keys();
                Iterator iter = sk.iterator();
                while(iter.hasNext()) {
                    SelectionKey key = (SelectionKey)iter.next();
                    key.channel().close();
                    key.cancel();
                }
            }
            selector.select(500);
            selector.close();
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("Aggregating Stats in " + _mode + " mode...");
        Aggregate agg = new Aggregate(ubc.Constants.serverFilePath, _mode, numWorkers, _time, startDate, new Date());
        System.out.println("Done!");
        System.exit(0);

    }

    // Accepts a new client
    public void accept(SelectionKey key) {
        try {
            ServerSocketChannel serverSocketChannel = (ServerSocketChannel) key.channel();
            SocketChannel client = serverSocketChannel.accept();
            client.configureBlocking(false);
            client.register(selector, SelectionKey.OP_READ, ByteBuffer.allocate(1));
            String hostname = client.socket().getInetAddress().getHostAddress();
            if ( !workerNames.contains(hostname) ) {
                workerNames.add( client.socket().getInetAddress().getHostAddress() );
                numWorkers++;
            }
        } catch (Exception e) {
            System.err.println("accept.Exception");
            //e.printStackTrace();
        }
    }

    // Sends some uncrawled hostnames to the client
    public void writeToClient(SelectionKey key, int message) {
        try {
            if(message==ubc.CrawlWrapper.QUIT) { // If we are sending a quit message
                CrawlWrapper cw = new CrawlWrapper(null, message);
                ubc.nio.writeIncludingLength(key, cw);
                ubc.nio.close(key);
            }
            else { // If we are not sending a quit message, then remove items from our uncrawledHost queue

                Queue q = qo.uncrawledHosts.dequeue(_numSend);

                // Wrap the queue with a message into a single object
                CrawlWrapper cw = new CrawlWrapper(q, message);

                // Send it
                ubc.nio.writeIncludingLength(key, cw);
                // Print some info
                System.err.println("  Sent " + q.size() + " hostnames to " + ((SocketChannel)key.channel()).socket().getInetAddress().getHostAddress() );
//                System.out.println("Sent Hostnames: " + q.toString());

                // Set our interest to read
//                key.interestOps(SelectionKey.OP_READ);
                ubc.nio.close(key);
            }
        } catch (Exception e) {
            System.err.println("writeToClient.Exception");
            //e.printStackTrace();
        }

    }

    // Reads from a channel associated with a client
    public void readFromClient(SelectionKey key) {
        ByteBuffer buf = (ByteBuffer)key.attachment();
        try {
            // If we haven't gotten the size of the entire data yet, then find out the size
            if ( buf.capacity() == 1 ) {
                buf = ByteBuffer.allocate(4);
                int totalRead = 0;
                long startTime = System.currentTimeMillis();
                while(!Thread.interrupted()) {
                    curTime = System.currentTimeMillis();
                    if( (curTime-startTime) > 2000) // If you can't send 4 bytes in 2 seconds, you are getting disconnected
                        return;
                    int numRead = ubc.nio.read(key, buf);
                    if(numRead == -1)
                        return;
                    totalRead += numRead;
                    if(totalRead == 4)
                        break;
                }
                buf.clear();
                int msgLength = buf.getInt();
                try {
                    buf = ByteBuffer.allocate(msgLength);
                } catch (Exception e) {
                    return;
                }
                buf.clear();
            }

            // Read some data
            int numRead = ubc.nio.read(key, buf);

            // Attach the newly updated buf back to the key
            key.attach(buf);

            // If we have read the entire data, then process it
            if( buf.position() == buf.capacity() ) {
                CrawlWrapper cw = (CrawlWrapper)ubc.nio.toObject(buf.array());
                List<CrawlResult> list = (List<CrawlResult>)cw.getObject();
                
                // Add all the new CrawlResults to our queue
//                System.err.print("Got Result For: [");
                for(int i=0; i<list.size(); i++) {
//                    System.err.print(list.get(i).getHostPort() + ", ");
                    qo.crawledResults.enqueue( list.get(i) );
                }
//                System.err.print("] \n");

                // Print some info
                System.err.println("  Got " + list.size() + " results from " + ((SocketChannel)key.channel()).socket().getInetAddress().getHostAddress() );
                buf = null;
                // Reset the attachment 
                key.attach(ByteBuffer.allocate(1));
//                buf = ByteBuffer.allocate(4);
//                buf.clear();
//                buf.putInt(list.size()); // Lets us know how many results we received from the worker
//                buf.clear();
//                key.attach(buf);
                // Set our intereset to write
                key.interestOps(SelectionKey.OP_WRITE);
            }

        } catch (Exception e) {
            System.err.println("readFromClient.Exception");
            //e.printStackTrace();
            ubc.nio.close(key);
        }
    }
    

}
